package com.tencent.sr.iris.activity.service.kafka.listener;

import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.Tracer;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.tencent.sr.iris.activity.interfaces.enums.ActivityEventTypeEnum;
import com.tencent.sr.iris.activity.interfaces.event.dto.ActivityEventDTO;
import com.tencent.sr.iris.activity.service.config.ActivityConfig;
import com.tencent.sr.iris.activity.service.event.EventExecutorContext;
import com.tencent.sr.iris.activity.service.event.EventStandardService;
import com.tencent.sr.iris.activity.service.event.handler.IrisEventHandler;
import com.tencent.sr.iris.activity.service.event.vo.DistributionOrderAggrVO;
import com.tencent.sr.iris.user.core.service.interfaces.enums.UserLevelEnum;
import com.tencent.sr.iris.user.core.service.interfaces.request.message.UserActivityLevelUpRequest;
import com.tencent.sr.iris.user.core.service.interfaces.request.message.UserSupervisorChangeRequest;
import com.tencent.sr.iris.user.core.service.interfaces.request.tag.SpecialLeaderTagMqRequest;
import com.tencent.sr.rmall.common.exception.retail.TRetailErrorException;
import com.tencent.sr.rmall.springbootstarter.kafka.KafkaMetaMessage;
import com.tencent.sr.rmall.springbootstarter.monitor.util.MonitorUtil;
import com.tencent.sr.rmall.springbootstarter.utils.JsonUtils;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.DependsOn;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@DependsOn("kafkaParamConfiguration")
public class UserEventListener {

    @Resource
    private EventStandardService eventStandardService;

    @Resource
    private EventExecutorContext eventExecutorContext;

    @Resource
    private ActivityConfig activityConfig;

    @KafkaListener(topics = "${kafka.topics.specialInvite}",
            groupId = "${retail.kafka.consumer.groupId.irisActivityService}",
            containerFactory = "kafkaListenerContainerFactory")
    public void handleSpecialInviteMsg(KafkaMetaMessage kafkaMetaMessage, Acknowledgment ack) {
        log.info("handleSpecialInviteMsg.handle msg: {}", JSON.toJSONString(kafkaMetaMessage));
        try {
            String jsonString = String.valueOf(kafkaMetaMessage.getData());
            SpecialLeaderTagMqRequest request = JsonUtils.parse(jsonString,
                    SpecialLeaderTagMqRequest.class);

            eventStandardService.handleSpecialInviteMsg(request);
        } catch (Exception e) {
            log.error("handleSpecialInviteMsg 消息处理异常，kafkaMetaMessage:{}", JSON.toJSONString(kafkaMetaMessage), e);
        }
        ack.acknowledge();
    }

    @KafkaListener(topics = "${kafka.topics.activityLevelUpTopic}",
            groupId = "${retail.kafka.consumer.groupId.irisActivityService}",
            containerFactory = "kafkaListenerContainerFactory")
    public void handleLevelUpMsg(KafkaMetaMessage kafkaMetaMessage, Acknowledgment ack) {
        log.info("activityLevelUpTopic.handle msg: {}", JSON.toJSONString(kafkaMetaMessage));
        try {
            String jsonString = String.valueOf(kafkaMetaMessage.getData());
            UserActivityLevelUpRequest request = JsonUtils.parse(jsonString,
                    UserActivityLevelUpRequest.class);
            if (UserLevelEnum.MEMBER.getValue().equals(request.getAfterLevelId())
                    && activityConfig.getLevelUpMemeberBlackList().contains(request.getUid())) {
                log.info("activityLevelUpTopic.handle member blackList: {}", jsonString);
                return;
            }
            eventStandardService.handleUserActivityLevelUpMsg(request);
        } catch (Exception e) {
            log.error("activityLevelUpTopic 消息处理异常，kafkaMetaMessage:{}", JSON.toJSONString(kafkaMetaMessage), e);
        }
        ack.acknowledge();
    }

    @KafkaListener(topics = "${kafka.topics.supervisorChange}",
            groupId = "${retail.kafka.consumer.groupId.irisActivityService}",
            containerFactory = "kafkaListenerContainerFactory")
    public void handleSupervisorChangeMsg(KafkaMetaMessage kafkaMetaMessage, Acknowledgment ack) {
        log.info("supervisorChange.handle msg: {}", JSON.toJSONString(kafkaMetaMessage));
        try {
            String jsonString = String.valueOf(kafkaMetaMessage.getData());
            UserSupervisorChangeRequest request = JsonUtils.parse(jsonString,
                    UserSupervisorChangeRequest.class);
            eventStandardService.handleUserSupervisorChangeMsg(request);
        } catch (Exception e) {
            log.error("supervisorChange 消息处理异常，kafkaMetaMessage:{}", JSON.toJSONString(kafkaMetaMessage), e);
        }
        ack.acknowledge();
    }

    @KafkaListener(topics = "${kafka.topics.distributionOrderTopic}",
            groupId = "${retail.kafka.consumer.groupId.irisActivityService}",
            containerFactory = "kafkaListenerContainerFactory")
    public void handleDistributionOrderMsg(KafkaMetaMessage kafkaMetaMessage, Acknowledgment ack) {
        log.info("handleDistributionOrderMsg.handle msg: {}", JSON.toJSONString(kafkaMetaMessage));
        try {
            String jsonString = String.valueOf(kafkaMetaMessage.getData());
            JSONObject jsonObject = JSON.parseObject(jsonString);
            if (jsonObject.containsKey("data")) {
                JSONObject data = jsonObject.getJSONObject("data");
                DistributionOrderAggrVO request = JsonUtils.parse(data.toJSONString(),
                        DistributionOrderAggrVO.class);
                eventStandardService.handleDistributionOrderMsg(request);
            }
        } catch (Exception e) {
            log.error("handleDistributionOrderMsg 消息处理异常，kafkaMetaMessage:{}", JSON.toJSONString(kafkaMetaMessage), e);
        }
        ack.acknowledge();
    }

    @KafkaListener(topics = "${kafka.topics.activityEventTopic}",
            groupId = "${retail.kafka.consumer.groupId.irisActivityService}",
            containerFactory = "kafkaListenerContainerFactory")
    public void handleStandardEventMsg(KafkaMetaMessage kafkaMetaMessage, Acknowledgment ack) throws Exception {
        long startTime = System.currentTimeMillis();
        Throwable throwable = null;
        log.info("standardEvent.handle msg: {}", JSON.toJSONString(kafkaMetaMessage));
        Entry entry = null;
        try {
            // 资源名可使用任意有业务语义的字符串，注意数目不能太多（超过 1K），超出几千请作为参数传入而不要直接作为资源名
            entry = SphU.entry("iris_activity_event_standardEvent_handle");
            String jsonString = String.valueOf(kafkaMetaMessage.getData());
            handleEvent(jsonString);
        } catch (BlockException ex) {
            throwable = ex;
            log.error("standardEvent.handle 消息处理,触发限流,等待消息重试");
            throw new TRetailErrorException("互动营销处理标准事件限流");
        } catch (Exception e) {
            throwable = e;
            log.error("standardEvent.handle 消息处理异常，kafkaMetaMessage:{}", JSON.toJSONString(kafkaMetaMessage), e);
            // 若需要配置降级规则，需要通过这种方式记录业务异常
            Tracer.traceEntry(e, entry);
            throw e;
        } finally {
            // 务必保证 exit，务必保证每个 entry 与 exit 配对
            if (entry != null) {
                entry.exit();
            }
            MonitorUtil.getSingleMonitorService().logTransaction("activity_event_handler", "kafka",
                    startTime, throwable);
        }

        ack.acknowledge();
    }

    @KafkaListener(topics = "${kafka.topics.activityLowerPriorityEventTopic}",
            groupId = "${retail.kafka.consumer.groupId.irisActivityService}",
            containerFactory = "kafkaListenerContainerFactory")
    public void handleStandardLowerPriorityEventMsg(KafkaMetaMessage kafkaMetaMessage, Acknowledgment ack)
            throws Exception {
        long startTime = System.currentTimeMillis();
        Throwable throwable = null;
        log.info("standardLowerPriorityEvent.handle msg: {}", JSON.toJSONString(kafkaMetaMessage));
        Entry entry = null;
        try {
            // 资源名可使用任意有业务语义的字符串，注意数目不能太多（超过 1K），超出几千请作为参数传入而不要直接作为资源名
            entry = SphU.entry("iris_activity_lower_priority_event_standardEvent_handle");
            String jsonString = String.valueOf(kafkaMetaMessage.getData());
            handleEvent(jsonString);
        } catch (BlockException ex) {
            throwable = ex;
            log.error("standardLowerPriorityEvent.handle 消息处理,触发限流,等待消息重试");
            throw new TRetailErrorException("互动营销处理低优先级标准事件限流");
        } catch (Exception e) {
            throwable = e;
            log.error("standardLowerPriorityEvent.handle 消息处理异常，kafkaMetaMessage:{}",
                    JSON.toJSONString(kafkaMetaMessage), e);
            // 若需要配置降级规则，需要通过这种方式记录业务异常
            Tracer.traceEntry(e, entry);
            throw e;
        } finally {
            // 务必保证 exit，务必保证每个 entry 与 exit 配对
            if (entry != null) {
                entry.exit();
            }
            MonitorUtil.getSingleMonitorService().logTransaction("activity_lower_priority_event_handler", "kafka",
                    startTime, throwable);
        }

        ack.acknowledge();
    }

    public void handleEvent(String jsonString) throws Exception {
        ActivityEventDTO request = null;
        try {
            request = JsonUtils.parse(jsonString, ActivityEventDTO.class);
        } catch (Exception e) {
            log.error("standardEvent.handle解析事件失败，忽略事件,event:{}", jsonString);
            return;
        }
        IrisEventHandler irisEventHandler = eventExecutorContext
                .getEventHandler(ActivityEventTypeEnum.getByCode(request.getEventType()));
        if (irisEventHandler != null) {
            irisEventHandler.execute(request);
            return;
        }
        log.error("standardEvent.handle没找到对应的事件处理器,event:{}", jsonString);
    }

    public static void main(String[] args) {
        ActivityEventDTO eventDTO = new ActivityEventDTO();
        eventDTO.setUid("180000099050992");
        eventDTO.setPid(null);
        eventDTO.setEventType("SHARETIME_CHANGE");
        eventDTO.setEventInfo(
                "{\"etlStamp\":\"20230329142003\",\"id\":3335,\"runTimeKey\":\"202303291420\",\"shareUvTo29\":2,\"shareVisitTime\":\"2023-03-29\",\"shareVvToEndOfMonth\":2,\"uid\":\"180000099050992\",\"updateTime\":\"2023-03-29 14:20:03\"}");
        eventDTO.setIdempotent("180000099050992_20230310142003_202303291420_2023-03-29");
        eventDTO.setEventSource("DATASOURCE_PUSH");
        eventDTO.setEventTime(1678377600000L);
        eventDTO.setCreateTime(1678429204042L);
        System.out.println(JSON.toJSON(eventDTO));

    }

}
